// Copyright 2022 Huawei Cloud Computing Technology Co., Ltd.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#include "CaeTcpSocket.h"
#include "../cae_common/CaeLogConfig.h"
#include "../cae_agent/CaeAgent.h"
#include <cerrno>
#include <cstring>
#include <cstdio>
#include <cstdlib>
#include <sys/socket.h>
#include <linux/in.h>
#include <linux/tcp.h>
#include <unistd.h>
using namespace std;
CaeTcpSocket::CaeTcpSocket()
{
    m_fd = -1;
    m_status = -1;
    m_localIp = 0;
    m_remoteIp = 0;
    m_localPort = 0;
    m_remotePort = 0;
    m_socketOption = 0;
    m_eventNotice = nullptr;
    pthread_mutex_init(&m_socketMutex, nullptr);
}

CaeTcpSocket::~CaeTcpSocket()
{
    m_status = -1;
    if (m_fd != -1) {
        shutdown(m_fd, SHUT_RDWR);
        close(m_fd);
    }
    m_fd = -1;
}

void CaeTcpSocket::CloseSocket()
{
    m_status = -1;

    if (m_fd != -1) {
        shutdown(m_fd, SHUT_RDWR);
        close(m_fd);
    }
    m_fd = -1;
}

int CaeTcpSocket::Send(void *pPkt, size_t size)
{
    int byte;
    int err = static_cast<int>(size);
    int socketFd = this->m_fd;
    if (pPkt == nullptr) {
        return SOCKET_SEND_FAIL_RETRY;
    }
    pthread_mutex_lock(&m_socketMutex);
    signal(SIGPIPE, SIG_IGN);
    while (true) {
        byte = ::send(socketFd, pPkt, size, MSG_NOSIGNAL);
        if (static_cast<int>(size) == byte) {
            pthread_mutex_unlock(&m_socketMutex);
            return byte;
        } else if ((byte >= 0) && (static_cast<int>(size) > byte)) {
            size -= static_cast<uint32_t>(byte);
            pPkt = (void *)((uint8_t *)pPkt + byte);
            continue;
        } else if (!((errno == EAGAIN) || (errno == EWOULDBLOCK) || (errno == EINTR))) {
            err = SOCKET_SEND_FAIL_DISCONN;
            LOGE("Socket Send error, errno=%d.", errno);
            m_status = SOCKET_STATUS_ERROR;
            break;
        } else if (m_fd == -1) {
            err = SOCKET_SEND_FAIL_DISCONN;
            LOGE("SocketFd error(fd = -1), errno=%d.", errno);
            CaeAgent::GetInstance().CloseClient();
            break;
        } else {
            LOGD("Socket Send error, errno=%d.", errno);
            break;
        }
    }
    pthread_mutex_unlock(&m_socketMutex);
    this->EventNotice();
    return err;
}

int CaeTcpSocket::Recv(void *pPkt, size_t size)
{
    uint32_t flag = m_socketOption & SOCKET_OPTION_BITSET_QUICKACK;
    int byte;
    int socketFd = this->m_fd;

    if (pPkt == nullptr) {
        return SOCKET_RECV_FAIL_RETRY;
    }
    byte = ::recv(socketFd, pPkt, size, 0);
    if (byte <= 0) {
        flag = 0;
        if (!((byte < 0) && ((errno == EAGAIN) || (errno == EWOULDBLOCK) || (errno == EINTR)))) {
            LOGE("Socket receive error, errno=%d.", errno);
            m_status = SOCKET_STATUS_ERROR;
            this->EventNotice();
            byte = SOCKET_RECV_FAIL_DISCONN;
        }
    }

    if (flag) {
        setsockopt(socketFd, IPPROTO_TCP, TCP_QUICKACK, (int[]) { 1 }, sizeof(int));
    }

    return byte;
}

void CaeTcpSocket::SetSocketTimeout(long sec, long usec)
{
    struct timeval timeout{};
    timeout.tv_sec = sec;
    timeout.tv_usec = usec;

    if (setsockopt(m_fd, SOL_SOCKET, SO_SNDTIMEO, &timeout, sizeof(timeout)) < 0) {
        LOGD("Set socket opt SO_SNDTIMEO failed, errno: (%d): %s.", errno, strerror(errno));
    }
    if (setsockopt(m_fd, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof(timeout)) < 0) {
        LOGD("Set socket opt SO_RCVTIMEO failed, errno: (%d): %s.", errno, strerror(errno));
    }
}
